import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Reducer;

import java.io.IOException;
import java.util.*;

public class FriendReduce extends Reducer<Text,Text,Text, Text> {
    @Override
    protected void reduce(Text key, Iterable<Text> values, Context context) throws IOException, InterruptedException {
      /***
       * A为key 存储 取a为key时的value B ,然后把B再作为key取判断当Key为B时的value为多少，得出的列表就是A的二度好友，要排除B本身
       * ***/
        Set<String> set = new HashSet<String>();
        //思路，key对应的value列表中，是可以互相推荐的，当key为B时，B有好友D,C,A，而D当然就可以通过B去认识C和A
        for(Text text : values){
            set.add(text.toString());
        }
        List<friendAB> friendABlist = new ArrayList<friendAB>();
        //因为存在只对应一个好友的情况
        if(set.size()>1){
            //先取一个友人A,它会成为本轮reduce的key
            for (String friendA : set) {
                for (String friendB : set) {
                    //去重留存，因为会存在a和b互相反过来的情况
                    friendAB friendab = new friendAB(friendB,friendA);
                    if (!friendA.equals(friendB)) {
                        friendABlist.add(new friendAB(friendA,friendB));
                    }
                    //查看相反的情况是否存在，存在则将这条数据移除
                    int aa = friendABlist.indexOf(friendab);
                    if( aa>=0){
                        friendABlist.remove(friendab);
                    }
                }
            }
        }
        //遍历输出
        for(friendAB fab :friendABlist){
            context.write(new Text(fab.fa),new Text(fab.fb));
        }
    }

}
class friendAB{
    String fa , fb;
    public friendAB(String a ,String b){
        fa = a;
        fb = b;
    }
    @Override
    public boolean equals(Object obj) {
        friendAB friendAB =(friendAB)obj;
        return this.fa.equals(friendAB.fa) && (this.fb.equals(friendAB.fb));
    }
}
